package com.dragon.concurrent.demo.threadpool;

import java.util.LinkedList;
import java.util.stream.IntStream;

public class SimpleThreadPool {
    private final int size;

    private final int taskQueueSize;

    private static final DiscardPolicy DEFAULT_DISCARD_POLICY = () -> {
        throw new DiscardException("discard the task");
    };

    private final DiscardPolicy discardPolicy;


    private static final int DEFAULT_SIZE = 10;

    private static final int DEFAULT_TASK_QUEUE_SIZE = 2000;

    private static final ThreadGroup THREAD_GROUP = new ThreadGroup("POOL_GROUP ");

    private static final String THREAD_PREFIX = "SIMPLE_THREAD_POOL-";

    private volatile boolean destroy = false;

    private static volatile int seq = 0;

    private static final LinkedList<Runnable> TASK_QUEUE = new LinkedList();

    private static final LinkedList<WorkTask> THREAD_QUEUE = new LinkedList();

    public SimpleThreadPool() {
        this(DEFAULT_SIZE, DEFAULT_TASK_QUEUE_SIZE, DEFAULT_DISCARD_POLICY);
    }

    public SimpleThreadPool(int size, int taskQueueSize, DiscardPolicy discardPolicy) {
        this.size = size;
        this.taskQueueSize = taskQueueSize;
        this.discardPolicy = discardPolicy;
        init();

    }

    private void init() {
        for (int i = 0; i < size; i++) {
            createWorkTask();
        }
    }

    private void createWorkTask() {
        WorkTask workTask = new WorkTask(THREAD_GROUP, THREAD_PREFIX + (seq++));
        workTask.start();
        THREAD_QUEUE.add(workTask);
    }

    public void shutdown() {
        while (!TASK_QUEUE.isEmpty()) {
            try {
                Thread.sleep(100L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        int initVal = THREAD_QUEUE.size();
        while (initVal > 0) {
            for (WorkTask workTask : THREAD_QUEUE) {
                if (workTask.taskState == TaskState.BLOCKED) {
                    workTask.interrupt();
                    workTask.close();
                    initVal--;
                } else {
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
        this.destroy = true;
        System.out.println("the threadPool disposed... ");
    }

    public void submit(Runnable runnable) {
        if (destroy) {
            throw new IllegalStateException("the threadPool is disposed, not allow submit task");
        }
        synchronized (TASK_QUEUE) {
            if (TASK_QUEUE.size() > this.taskQueueSize) {
                discardPolicy.discard();
            }
            TASK_QUEUE.add(runnable);
            TASK_QUEUE.notifyAll();
        }
    }

    public int getSize() {
        return this.size;
    }

    public int getTaskQueueSize() {
        return this.taskQueueSize;
    }

    public boolean destroy() {
        return this.destroy;
    }

    public static class DiscardException extends RuntimeException {

        public DiscardException(String message) {
            super(message);
        }

    }

    public interface DiscardPolicy {
        void discard() throws DiscardException;
    }


    /**
     * 自定义线程
     */
    private static class WorkTask extends Thread {
        private volatile TaskState taskState = TaskState.FREE;

        public WorkTask(ThreadGroup group, String name) {
            super(group, name);
        }

        @Override
        public void run() {
            OUTER:
            while (this.taskState != TaskState.DEAD) {
                synchronized (TASK_QUEUE) {
                    while (TASK_QUEUE.isEmpty()) {
                        try {
                            taskState = TaskState.BLOCKED;
                            TASK_QUEUE.wait();
                        } catch (InterruptedException e) {
                            break OUTER;
                        }
                    }
                    Runnable runnable = TASK_QUEUE.removeFirst();
                    if (runnable != null) {
                        taskState = TaskState.RUNNABLE;
                        runnable.run();
                        taskState = TaskState.FREE;
                    }
                }
            }
        }

        public TaskState getTaskState() {
            return this.taskState;
        }

        public void close() {
            this.taskState = TaskState.DEAD;
        }
    }

    /**
     * 工作线程的状态
     */
    private enum TaskState {
        FREE, RUNNABLE, BLOCKED, DEAD;
    }

    public static void main(String[] args) {
        SimpleThreadPool simpleThreadPool = new SimpleThreadPool();
        IntStream.range(0, 400)
                .forEach(i -> {
                    simpleThreadPool.submit(() -> {
                        System.out.println("the runnable [" + i + "] do serviced by " + Thread.currentThread() + "starting");
//                       try {
//                           Thread.sleep(100L);
//                       } catch (InterruptedException e) {
//                           e.printStackTrace();
//                       }
                        System.out.println("the runnable [" + i + "] do serviced by " + Thread.currentThread() + " finished");
                    });
                });
        simpleThreadPool.shutdown();
        simpleThreadPool.submit(() -> {
            System.out.println("=================");
        });

    }
}
